/*
 * Copyright (C) 2007 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.google.common.util.concurrent;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.AbstractExecutorService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Factory and utility methods for {@link java.util.concurrent.Executor},
 * {@link ExecutorService}, and {@link ThreadFactory}.
 * 
 * @author Eric Fellheimer
 * @author Kyle Littlefield
 * @author Justin Mahoney
 * @since 2009.09.15 <b>tentative</b>
 */
public class Executors {

    /**
     * Converts the given ThreadPoolExecutor into an ExecutorService that exits
     * when the application is complete. It does so by using daemon threads and
     * adding a shutdown hook to wait for their completion.
     * 
     * <p>
     * This is mainly for fixed thread pools. See
     * {@link java.util.concurrent.Executors#newFixedThreadPool(int)}.
     * 
     * @param executor
     *            the executor to modify to make sure it exits when the
     *            application is finished
     * @param terminationTimeout
     *            how long to wait for the executor to finish before terminating
     *            the JVM
     * @param timeUnit
     *            unit of time for the time parameter
     * @return an unmodifiable version of the input which will not hang the JVM
     */
    public static ExecutorService getExitingExecutorService(
            ThreadPoolExecutor executor, long terminationTimeout,
            TimeUnit timeUnit) {
        executor.setThreadFactory(daemonThreadFactory(executor
                .getThreadFactory()));

        ExecutorService service = java.util.concurrent.Executors
                .unconfigurableExecutorService(executor);

        addDelayedShutdownHook(service, terminationTimeout, timeUnit);

        return service;
    }

    /**
     * Converts the given ScheduledThreadPoolExecutor into a
     * ScheduledExecutorService that exits when the application is complete. It
     * does so by using daemon threads and adding a shutdown hook to wait for
     * their completion.
     * 
     * <p>
     * This is mainly for fixed thread pools. See
     * {@link java.util.concurrent.Executors#newScheduledThreadPool(int)}.
     * 
     * @param executor
     *            the executor to modify to make sure it exits when the
     *            application is finished
     * @param terminationTimeout
     *            how long to wait for the executor to finish before terminating
     *            the JVM
     * @param timeUnit
     *            unit of time for the time parameter
     * @return an unmodifiable version of the input which will not hang the JVM
     */
    public static ScheduledExecutorService getExitingScheduledExecutorService(
            ScheduledThreadPoolExecutor executor, long terminationTimeout,
            TimeUnit timeUnit) {
        executor.setThreadFactory(daemonThreadFactory(executor
                .getThreadFactory()));

        ScheduledExecutorService service = java.util.concurrent.Executors
                .unconfigurableScheduledExecutorService(executor);

        addDelayedShutdownHook(service, terminationTimeout, timeUnit);

        return service;
    }

    /**
     * Add a shutdown hook to wait for thread completion in the given
     * {@link ExecutorService service}. This is useful if the given service uses
     * daemon threads, and we want to keep the JVM from exiting immediately on
     * shutdown, instead giving these daemon threads a chance to terminate
     * normally.
     * 
     * @param service
     *            ExecutorService which uses daemon threads
     * @param terminationTimeout
     *            how long to wait for the executor to finish before terminating
     *            the JVM
     * @param timeUnit
     *            unit of time for the time parameter
     */
    public static void addDelayedShutdownHook(final ExecutorService service,
            final long terminationTimeout, final TimeUnit timeUnit) {
        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
            public void run() {
                try {
                    // We'd like to log progress and failures that may arise in
                    // the
                    // following code, but unfortunately the behavior of logging
                    // is undefined in shutdown hooks.
                    // This is because the logging code installs a shutdown hook
                    // of its
                    // own. See Cleaner class inside {@link LogManager}.
                    service.shutdown();
                    service.awaitTermination(terminationTimeout, timeUnit);
                } catch (InterruptedException ignored) {
                    // We're shutting down anyway, so just ignore.
                }
            }
        }));
    }

    /**
     * Converts the given ThreadPoolExecutor into an ExecutorService that exits
     * when the application is complete. It does so by using daemon threads and
     * adding a shutdown hook to wait for their completion.
     * 
     * <p>
     * This method waits 120 seconds before continuing with JVM termination,
     * even if the executor has not finished its work.
     * 
     * <p>
     * This is mainly for fixed thread pools. See
     * {@link java.util.concurrent.Executors#newFixedThreadPool(int)}.
     * 
     * @param executor
     *            the executor to modify to make sure it exits when the
     *            application is finished
     * @return an unmodifiable version of the input which will not hang the JVM
     */
    public static ExecutorService getExitingExecutorService(
            ThreadPoolExecutor executor) {
        return getExitingExecutorService(executor, 120, TimeUnit.SECONDS);
    }

    /**
     * Converts the given ThreadPoolExecutor into a ScheduledExecutorService
     * that exits when the application is complete. It does so by using daemon
     * threads and adding a shutdown hook to wait for their completion.
     * 
     * <p>
     * This method waits 120 seconds before continuing with JVM termination,
     * even if the executor has not finished its work.
     * 
     * <p>
     * This is mainly for fixed thread pools. See
     * {@link java.util.concurrent.Executors#newScheduledThreadPool(int)}.
     * 
     * @param executor
     *            the executor to modify to make sure it exits when the
     *            application is finished
     * @return an unmodifiable version of the input which will not hang the JVM
     */
    public static ScheduledExecutorService getExitingScheduledExecutorService(
            ScheduledThreadPoolExecutor executor) {
        return getExitingScheduledExecutorService(executor, 120,
                TimeUnit.SECONDS);
    }

    /**
     * Returns a {@link ThreadFactory} which creates daemon threads. This is
     * implemented by wrapping
     * {@link java.util.concurrent.Executors#defaultThreadFactory()}, marking
     * all new threads as daemon threads
     * 
     * @return a {@link ThreadFactory} which creates daemon threads
     */
    public static ThreadFactory daemonThreadFactory() {
        return daemonThreadFactory(java.util.concurrent.Executors
                .defaultThreadFactory());
    }

    /**
     * Wraps another {@link ThreadFactory}, making all new threads daemon
     * threads.
     * 
     * @param factory
     *            the {@link ThreadFactory} used to generate new threads
     * @return a new {@link ThreadFactory} backed by {@code factory} whose
     *         created threads are all daemon threads
     */
    public static ThreadFactory daemonThreadFactory(ThreadFactory factory) {
        return new DaemonThreadFactory(factory);
    }

    /**
     * Creates an executor service that runs each task in the thread that
     * invokes {@code execute/submit}, as in {@link CallerRunsPolicy} This
     * applies both to individually submitted tasks and to collections of tasks
     * submitted via {@code invokeAll} or {@code invokeAny}. In the latter case,
     * tasks will run serially on the calling thread. Tasks are run to
     * completion before a {@code Future} is returned to the caller (unless the
     * executor has been shutdown).
     * 
     * <p>
     * Although all tasks are immediately executed in the thread that submitted
     * the task, this {@code ExecutorService} imposes a small locking overhead
     * on each task submission in order to implement shutdown and termination
     * behavior.
     * 
     * <p>
     * The implementation deviates from the {@code ExecutorService}
     * specification with regards to the {@code shutdownNow} method. First,
     * "best-effort" with regards to canceling running tasks is implemented as
     * "no-effort". No interrupts or other attempts are made to stop threads
     * executing tasks. Second, the returned list will always be empty, as any
     * submitted task is considered to have started execution. This applies also
     * to tasks given to {@code invokeAll} or {@code invokeAny} which are
     * pending serial execution, even the subset of the tasks that have not yet
     * started execution. It is unclear from the {@code ExecutorService}
     * specification if these should be included, and it's much easier to
     * implement the interpretation that they not be. Finally, a call to
     * {@code shutdown} or {@code shutdownNow} may result in concurrent calls to
     * {@code invokeAll/invokeAny} throwing RejectedExecutionException, although
     * a subset of the tasks may already have been executed.
     */
    public static ExecutorService sameThreadExecutor() {
        return new SameThreadExecutorService();
    }

    // See sameThreadExecutor javadoc for behavioral notes.
    private static class SameThreadExecutorService extends
            AbstractExecutorService {

        /**
         * Lock used whenever accessing the state variables (runningTasks,
         * shutdown, terminationCondition) of the executor
         */
        private final Lock lock = new ReentrantLock();

        /** Signaled after the executor is shutdown and running tasks are done */
        private final Condition termination = lock.newCondition();

        /*
         * Conceptually, these two variables describe the executor being in one
         * of three states: - Active: shutdown == false - Shutdown: runningTasks
         * > 0 and shutdown == true - Terminated: runningTasks == 0 and shutdown
         * == true
         */
        private int runningTasks = 0;
        private boolean shutdown = false;

        /* @Override */
        public void execute(Runnable command) {
            startTask();
            try {
                command.run();
            } finally {
                endTask();
            }
        }

        /* @Override */
        public boolean isShutdown() {
            lock.lock();
            try {
                return shutdown;
            } finally {
                lock.unlock();
            }
        }

        /* @Override */
        public void shutdown() {
            lock.lock();
            try {
                shutdown = true;
            } finally {
                lock.unlock();
            }
        }

        // See sameThreadExecutor javadoc for unusual behavior of this method.
        /* @Override */
        public List<Runnable> shutdownNow() {
            shutdown();
            return Collections.emptyList();
        }

        /* @Override */
        public boolean isTerminated() {
            lock.lock();
            try {
                return shutdown && runningTasks == 0;
            } finally {
                lock.unlock();
            }
        }

        /* @Override */
        public boolean awaitTermination(long timeout, TimeUnit unit)
                throws InterruptedException {
            long nanos = unit.toNanos(timeout);
            lock.lock();
            try {
                for (;;) {
                    if (isTerminated()) {
                        return true;
                    } else if (nanos <= 0) {
                        return false;
                    } else {
                        nanos = termination.awaitNanos(nanos);
                    }
                }
            } finally {
                lock.unlock();
            }
        }

        /**
         * Checks if the executor has been shut down and increments the running
         * task count.
         * 
         * @throws RejectedExecutionException
         *             if the executor has been previously shutdown
         */
        private void startTask() {
            lock.lock();
            try {
                if (isShutdown()) {
                    throw new RejectedExecutionException(
                            "Executor already shutdown");
                }
                runningTasks++;
            } finally {
                lock.unlock();
            }
        }

        /**
         * Decrements the running task count.
         */
        private void endTask() {
            lock.lock();
            try {
                runningTasks--;
                if (isTerminated()) {
                    termination.signalAll();
                }
            } finally {
                lock.unlock();
            }
        }
    }
}
